/******************************************************************************
* Copyright (C) 2020 ShenZhen Powerdata Information Technology Co.,Ltd
* All Rights Reserved.
* 本软件为***研制。未经本公司正式书面同意，其他任何个人、团体不得使用、
* 复制、修改或发布本软件.
*****************************************************************************/

package com.xxl.job.listener.config;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import com.xxl.job.listener.service.MailService;

import lombok.extern.slf4j.Slf4j;

/**
* @Title:
* @ClassName:  ConsumerThread   
* @Description:TODO  
* @author: xgf
* @date:   2020年2月23日 下午4:50:36   
* @since   JDK1.8
* @history 2020年2月23日 TODO 新建
*/
@Slf4j
public class ConsumerThread implements Runnable{

	
	private  RocketMqConfig rocketMqConfig;
	
	private MailService mailService;
	
    private DefaultMQPushConsumer consumer;
	
	public ConsumerThread(RocketMqConfig rocketMqConfig, MailService mailService) {
		this.rocketMqConfig = rocketMqConfig;
		this.mailService = mailService;
	}



	@Override
	public void run() {
		
		consumer = new DefaultMQPushConsumer(rocketMqConfig.getConsumerName());
		consumer.setNamesrvAddr(rocketMqConfig.getNamesrv());
		
		//需要添加 重试次数 限制
		//需要添加每次只能消费一条
		try {
			consumer.subscribe(rocketMqConfig.getTopic(), "*");
		} catch (MQClientException e1) {
			log.error(e1.toString());
		}
		consumer.registerMessageListener(new MessageListenerConcurrently() {
			
			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
				ConsumeConcurrentlyStatus consumeMessage = mailService.consumeMessage(msgs, context);
				return consumeMessage;
			}
		});
		
		try {
			consumer.start();
		} catch (MQClientException e) {
			log.error(e.toString());
		}
	}

}
